package com.redis.demo.listen;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

import org.apache.commons.lang.StringUtils;

import com.redis.demo.Message;
import com.redis.demo.User;
import com.redis.demo.utils.MessageType;
import com.redis.demo.utils.SerializeUtils;

import redis.clients.jedis.JedisPubSub;

/**
 * 监听到订阅模式接受到消息时的回调 (onPMessage) 监听到订阅频道接受到消息时的回调 (onMessage ) 订阅频道时的回调(
 * onSubscribe ) 取消订阅频道时的回调( onUnsubscribe ) 订阅频道模式时的回调 ( onPSubscribe )
 * 取消订阅模式时的回调( onPUnsubscribe )
 * 
 * @author wesley
 *
 */
public class Listen extends JedisPubSub {

	@Override
	public void onMessage(String channel, String message) {
		if (StringUtils.isNotBlank(message)) {
			Message msg = (Message) SerializeUtils.unSerialize2(message);
			System.out.println("------message-----" + msg);
			if (msg.getType().equals(MessageType.text)) {
				System.err.println("-----text message----"); // 文本消息
			}
			if (msg.getType().equals(MessageType.object)) {
				System.err.println("-----object message----");// 对象消息，在这里可以处理一些基本的业务。
			}
			if (msg.getType().equals(MessageType.queue)) {
				System.err.println("-----queue message----"); // 队列消息，可以执行大批量的业务，使用多线程处理。
				final BlockingQueue linkQueue = (BlockingQueue) msg.getMessage();
				System.err.println("----待处理的消息数----" + linkQueue.size());
				final int size = linkQueue.size();
				//这里使用线程是为了解决数据大的时候 这里可以改用线程池做。
				Thread t = new Thread(new Runnable() {
					public void run() {
						for (int i = 0; i <= size; i++) {
							try {
								System.err.println("----取出队列消息---" + i + linkQueue.take());
							} catch (InterruptedException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							}
						}
					}
				});
				t.start();
			}
		}
	}

	@Override
	public void onPMessage(String pattern, String channel, String message) {
		// TODO Auto-generated method stub
		System.out.println(pattern + "," + channel + "," + message);

	}

	@Override
	public void onSubscribe(String channel, int subscribedChannels) {
		// TODO Auto-generated method stub
		System.out.println("onSubscribe: channel[" + channel + "]," + "subscribedChannels[" + subscribedChannels + "]");
	}

	@Override
	public void onUnsubscribe(String channel, int subscribedChannels) {
		// TODO Auto-generated method stub
		System.out.println(
				"onUnsubscribe: channel[" + channel + "], " + "subscribedChannels[" + subscribedChannels + "]");
	}

	@Override
	public void onPUnsubscribe(String pattern, int subscribedChannels) {
		// TODO Auto-generated method stub
		System.out.println("onPUnsubscribe: pattern[" + pattern + "]," +

				"subscribedChannels[" + subscribedChannels + "]");

	}

	@Override
	public void onPSubscribe(String pattern, int subscribedChannels) {
		System.out.println("onPSubscribe: pattern[" + pattern + "], " +

				"subscribedChannels[" + subscribedChannels + "]");

	}

}
